package com.example.hxds.snm.task;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.map.MapUtil;
import com.example.hxds.common.exception.HxdsException;
import com.example.hxds.snm.entity.NewOrderMessage;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author XUYANG
 * @Date 2024-03-20 15:22
 */
@Component
@Slf4j
public class NewOrderMassageTask {
    @Resource
    private ConnectionFactory connectionFactory;
    public void sendNewOrderMessage(ArrayList<NewOrderMessage> list){
        int ttl=16*60*1000;
        String exchangeName="new_order_private";
        try (Connection connection=connectionFactory.newConnection();
             Channel channel=connection.createChannel(); ){
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            HashMap param=new HashMap();

            for (NewOrderMessage message:list){
                HashMap map=new HashMap(){{
                    put("orderId",message.getOrderId());
                    put("from",message.getFrom());
                    put("to",message.getTo());
                    put("expectsFee",message.getExpectsFee());
                    put("mileage",message.getMileage());
                    put("minute",message.getMinute());
                    put("distance",message.getDistance());
                    put("favourFee",message.getFavourFee());
                }};
                AMQP.BasicProperties properties=new AMQP.BasicProperties()
                        .builder().contentEncoding("UTF-8").headers(map)
                        .expiration(ttl+"").build();
                String queueName="queue_"+message.getUserId();
                String routingKey=message.getUserId();
                channel.queueDeclare(queueName,true,false,false,param);
                channel.queueBind(queueName,exchangeName,routingKey);
                channel.basicPublish(exchangeName,routingKey,properties,("新订单"+message.getOrderId()).getBytes());
                log.debug(message.getUserId()+"的订单消息发送成功");
                System.out.println(message.getUserId()+"的订单消息发送成功");
            }
        }catch (Exception e){
            log.error("消息队列执行异常",e);
            throw new HxdsException("订单消息发送失败");
        }
    }
    @Async
    public void sendNewOrderMessageAsync(ArrayList<NewOrderMessage> list){
        this.sendNewOrderMessage(list);
    }
    public List<NewOrderMessage> receiveNewOrderMessage(long userId){
        String exchangeName="new_order_private";
        String queueName="queue_"+userId;
        String routingKey=userId+"";

        List<NewOrderMessage> res=new ArrayList<>();
        try (Connection connection=connectionFactory.newConnection();
             Channel channel =connection.createChannel()){
            //定义交换机和交换机模式
            channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT);
            //声明队列（队列名，持久化缓存消息，消息接收不加锁，消息全部接受完不删除队列）
            channel.queueDeclare(queueName,true,false,false,null);
            //绑定要接收的队列
            channel.queueBind(queueName,exchangeName,routingKey);
            //避免一次接受过多消息进行限流，每次接收10条，然后循环接收
            channel.basicQos(0,10,true);
            while (true){
                GetResponse response=channel.basicGet(queueName,false);
                if (response!=null){
                    AMQP.BasicProperties properties=response.getProps();
                    Map<String,Object> map=properties.getHeaders();
                    String orderId = MapUtil.getStr(map, "orderId");
                    String from = MapUtil.getStr(map, "from");
                    String to = MapUtil.getStr(map, "to");
                    String expectsFee = MapUtil.getStr(map, "expectsFee");
                    String mileage = MapUtil.getStr(map, "mileage");
                    String minute = MapUtil.getStr(map, "minute");
                    String distance = MapUtil.getStr(map, "distance");
                    String favourFee = MapUtil.getStr(map, "favourFee");

                    NewOrderMessage message = new NewOrderMessage();
                    message.setOrderId(orderId);
                    message.setFrom(from);
                    message.setTo(to);
                    message.setExpectsFee(expectsFee);
                    message.setMileage(mileage);
                    message.setMinute(minute);
                    message.setDistance(distance);
                    message.setFavourFee(favourFee);

                    res.add(message);

                    byte[] body = response.getBody();
                    String msg=new String(body);
                    log.debug("从RabbitMQ中接收到的消息为："+msg);
                    System.out.println("从RabbitMQ中接收到的消息为："+msg);
                    long deliveryTag=response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag,false);
                }else {
                    //System.out.println("从RabbitMQ中未接收到消息");
                    break;
                }
            }
            ListUtil.reverse(res);
            return res;
        }catch (Exception e){
            log.error("接收消息执行异常",e);
            throw new HxdsException("接收新订单失败");
        }
    }
    public void deleteNewOrderQueue(long userId){
        String exchangeName="new_order_private";
        String queueName="queue_"+userId;
        try (Connection connection=connectionFactory.newConnection();
             Channel channel =connection.createChannel();){
            channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT);
            channel.queueDelete(queueName);
            log.debug(userId+"的队列删除成功");
            System.out.println(userId+"的队列删除成功");
        }catch (Exception e){
            log.error(userId+"的订单队列删除失败");
            throw new HxdsException("订单队列删除失败");
        }
    }
    @Async
    public void deleteNewOrderQueueAsync(long userId){
        this.deleteNewOrderQueue(userId);
    }
    public void clearNewOrderQueue(long userId){
        String exchangeName="new_order_private";
        String queueName="queue_"+userId;
        try (Connection connection=connectionFactory.newConnection();
             Channel channel =connection.createChannel();){
            channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT);
            channel.queuePurge(queueName);
            log.debug(userId+"的队列清空删除成功");
        }catch (Exception e){
            log.debug(userId+"的队列清空删除失败");
            throw new HxdsException(userId+"队列清空删除失败");
        }
    }
    @Async
    public void clearNewOrderQueueAsync(long userId){
        this.clearNewOrderQueue(userId);
    }
}
